/*
 * ===================================================================
 *  
 *      Filename:  log_client_daemon.c
 *
 *   Description:  This file is the main routine for the msg manipulation.
 *                 All of the other modules are loaded by the daemon, and
 *                 the daemon thread is waiting on the shared memory.
 *
 *       Version:  1.0
 *        Create:  2010/07/13
 *
 *        Author:  Keven Zhu     kevenzhu@tencent.com
 *     Copyright:  Reserved by Tencent.com Co.ltd
 *          
 * ====================================================================
 */
#include <unistd.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <errno.h>
#include <fcntl.h>
#include <signal.h>
#include <time.h>
#include <pthread.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/msg.h>

#include "lib/util.h"
#include "log_client_api.h"
#include "log_client_queue.h"

#define LOCKFILE "/var/run/hplog.pid"   /*file lock which used to keep the single instance*/
#define LOCKMODE (S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH)

pthread_t daemon_pid = 0;
pthread_t policy_pid = 0;
char*  shm_addr = NULL;
int    shm_id    = INV_HANDLER;
int    msg_queue = INV_HANDLER; 

int global_log_level = 0;

/*local share memory header for the local process*/
local_share_mem_ctl_header daemon_ctl_header;

#if IPC_MSG
struct log_msg
{
    long msg_type;
    char text[MAX_LOG_MSG_SIZE];
};
#endif

/* set the log level for the daemon process*/
int set_daemon_log_level(int log_level)
{
	/*8 means forbid all the msg*/
	if(log_level <0 || log_level >8)
	{
		return NOERR;
	}
	else
	{
		global_log_level = log_level;
		return NOERR;
	}
}

inline int update_destination_due_to_network()
{
    #define MAX_FAILURE_COUNTER 10
    static time_t seconds = 0;
    static int    failure_num = 0;
    time_t cur_time = time((time_t*)NULL);
    if((cur_time - seconds > 3))// || (failure_num++ > MAX_FAILURE_COUNTER))
    {
        failure_num = 0;
        seconds = cur_time;
        //remove_srv_from_list(get_destination());
        update_destination_addr(SEARCH_EXT);
        log_doctor(doctor_level,"new server is %02x\n",get_destination());
        reinit_protocol_tcp(get_destination());
        reinit_protocol_udp(get_destination());
        return NOERR;
    }
    else
    {
        return NOERR;
    }
}

void* hplog_client_daemon(void *argv)
{
    int iret   = 0,log_level = 0;
    #if IPC_SHM
    char* buff = NULL;
    #endif
	#if IPC_MSG
    struct log_msg msg_data;
    char*  buff= msg_data.text;
    #endif

    log_doctor(doctor_level,"The daemon process is running\n");
    while(1)
    {
	    #if IPC_MSG
        if(msgrcv(msg_queue,(void*)&msg_data,MAX_LOG_MSG_SIZE,0,0) != -1)
        {
        #endif
        #if IPC_SHM
		/*wait on the share memory*/
        hplog_sem_reader_wait();
        buff = dequeue(&daemon_ctl_header);
        /*increase the counter for the writer*/
        hplog_sem_wirter_post();
        while(buff)
        {
        #endif
            iret      = *((short*)buff);
            iret = iret - 1;/*sub the log level len*/
            *((short*)buff) = iret;
            log_level = *((char*)(buff+iret));
			/*test the log_level again*/
			if(log_level<global_log_level)
			{
				continue;
			}
            /*test if we need to do the color*/
            if(COLOR_ENABLED == get_color_status())
            {
                color_msg(buff);
            }
            /*if the log level is greater than WARN,we will send it with TCP,otherwise with UDP*/
            if(log_level > WARN)
            {
                /*EROR CRIT ALET EMGT TCP*/
                if(send_log_via_tcp(buff))//error to send the packet,try to change the srv
                {
                    /*update the srv list with default srv*/
                    log_doctor(doctor_level,"failed to send the packet with tcp using %02x\n",get_destination());
                    update_destination_due_to_network();
                }
                else
                {
                    if(get_backup_status())
                    {
                        flush_backup_to_srv_with_tcp();
                    }
                }
            }
            else
            {
                /*DEBG INFO NOTC WARN UDP*/
                if(send_log_via_udp(buff))//error to send the packet,try to change the srv
                {
                    /*update the srv list with default srv*/
                    log_doctor(doctor_level,"failed to send the packet with udp using %02x\n",get_destination());
                    update_destination_due_to_network();
                }
                else
                {
                    if(get_backup_status())
                    {
                        flush_backup_to_srv_with_tcp();
                    }
                }
            }
            #if IPC_SHM
            /*wait on the share memory*/
            hplog_sem_reader_wait();
            buff = dequeue(&daemon_ctl_header);
            /*increase the counter for the writer*/
            hplog_sem_wirter_post();
            #endif
        }
    }
    return NULL;
}

/* split the share memory to the queue*/
int split_share_memory()
{
    int iret = 0;

    share_mem_ctl_header* header = (share_mem_ctl_header*)shm_addr;

    /*set the memory of control to zero*/
    //memset((void*)header,0,sizeof(share_mem_ctl_header));
    init_queue(header,MAX_LOG_MSG_COUNTER);

    /*map the local header to the share memory header*/
    memset((void*)&daemon_ctl_header,0,sizeof(local_share_mem_ctl_header));
    daemon_ctl_header.max_queue      = &(header->max_queue);
    daemon_ctl_header.max_queue_mask = &(header->max_queue_mask);
    daemon_ctl_header.enqueue_counter= &(header->enqueue_counter);
    daemon_ctl_header.dequeue_counter= &(header->dequeue_counter);


    /*assign the share memory address to the log_list*/
    while(iret<MAX_LOG_MSG_COUNTER)
    {
        daemon_ctl_header.log_list[iret] = (char*)(shm_addr + sizeof(share_mem_ctl_header) + MAX_LOG_MSG_SIZE*iret); 
        iret ++;        
    }
    return NOERR;
}

/* initialze the share memory
 * 
 * return value:
 * positive means failure,error could be found in the head file or with the help of plogerr();
 * 0 means ok(launched)*/
int initialize_share_memory()
{    
    /*share memory map*/
    /*|----------------------------------------------------------------------------------------
     *|      management header     |       msg      |        msg        |           msg       |            
     *|----------------------------------------------------------------------------------------
     *|         msg        |            msg        |         msg       |         ...          |
     *|---------------------------------------------------------------------------------------*/
    int share_mem_size = sizeof(share_mem_ctl_header)+ MAX_LOG_MSG_COUNTER*MAX_LOG_MSG_SIZE;

    key_t key = (key_t)HPLOG_SHMM_KEY;
    /*create the share memory handler and map to our process*/
    shm_id = shmget(key,share_mem_size,IPC_CREAT|S_IRWXU|S_IRWXG|S_IRWXO);
    if(-1 == shm_id)
    {
        log_doctor(doctor_level,"failed to create the shared memory\n");
        return FITSM;                   
    }

    /*map the memory to the process*/
    shm_addr = (char*)shmat(shm_id,0,0);
    if(-1 == (long)shm_addr)
    {
        log_doctor(doctor_level,"failed to map the shm to our process\n");
        return FMPMM;
    }
    return split_share_memory();
}

int finitialize_share_memory()
{
    shmdt(shm_addr);
    shmctl(shm_id,IPC_RMID,0);
    return NOERR;
}

/* destroy the log daemon modules,Don't try to use,we will keep the daemon running,haha~~
 * 
 * return value:
 * positive means failure,error could be found in the head file or with the help of plogerr();
 * 0 means ok*/
int unload_hplog_client_deamon()
{
    log_doctor(doctor_level,"\n*****************************************************\n");
    log_doctor(doctor_level,"The hplog client daemon is ready to be killed  :( \n");
    log_doctor(doctor_level,"Policy thread is shutting down\n");
    pthread_cancel(policy_pid);
    log_doctor(doctor_level,"Daemon thread is shitting down\n");
    pthread_cancel(daemon_pid);
    
    log_doctor(doctor_level,"Policy module is unmounting\n");
    finit_log_policy();
    log_doctor(doctor_level,"UDP module is unmounting\n");
    finit_protocol_udp();
    log_doctor(doctor_level,"TCP module is unmounting\n");
    finit_protocol_tcp();
    log_doctor(doctor_level,"Color  module is unmounting\n");
    finit_log_color();
    log_doctor(doctor_level,"Backup module is unmounting\n");
    finit_log_backup();
    log_doctor(doctor_level,"Loadbalance module is unmounting\n");
    finit_loadbalance();
    #if IPC_MSG
    log_doctor(doctor_level,"Msg queue module is unmounting\n");
    msgctl(msg_queue,IPC_RMID,0);
    #endif
	#if IPC_SHM
    log_doctor(doctor_level,"Mutex module is unmounting\n");
    finitialize_hplog_mutex_sem();
    log_doctor(doctor_level,"Share memory module is unmounting\n");
    finitialize_share_memory();
    #endif
    log_doctor(doctor_level,"LOG doctor module is unmounting\n");
    finit_log_doctor();
    log_doctor(doctor_level,"*****************************************************\n");
    return NOERR;
}

/*hanlder signal for the kill*/
void handler_signal(int signo)
{
    unload_hplog_client_deamon();
    exit(1);
}

/* initialize the hplog client daemon
 * 
 * return value:
 * positive means failure,error could be found in the head file or with the help of plogerr();
 * 0 means ok(launched)*/
int initialize_hplog_daemon(int policy_fd)
{
    int iret             = 0;

	#if IPC_MSG
    /*the share memory solution is abandoned,we will msg queue instead*/
    msg_queue = msgget((key_t)HPLOG_MSG_QUEUE,0666|IPC_CREAT);
    if(msg_queue == INV_HANDLER)
    {
        log_doctor(doctor_level,"failed to initilize the msg queue\n");
        iret = NOMEM;
        goto FSHMM;
    }
	#endif
    #if IPC_SHM
    iret = initialize_share_memory();
    if(iret)
    {
        goto FSHMM;
    }

    iret = initialize_hplog_mutex_sem(DAEMON_PROCESS);
    if(iret)
    {
        goto FSHMM;
    }
    #endif

    iret = init_log_color();
    if(iret)
    {
        #if IPC_MSG	
        goto FITMG;
        #endif
        #if IPC_SHM
        goto FISEM;
        #endif				
    }
    /*initialize the loadbalance modules*/
    init_loadbalance();
    if(update_destination_addr(SEARCH_ALL))
    {
        /*failed to load the lb modules*/
        iret = NOMEM;
        goto FICLR;
    }

    /*initialize the backup modules*/
    iret = init_log_backup();
    if(iret)
    {
        goto FILBM;
    }

    /*initialize the tcp modules*/
    iret = init_protocol_tcp();
    if(iret)
    {
        goto FIBKM;
    }

    /*initialzie the udp modules*/
    iret = init_protocol_udp();
    if(iret)
    {
        goto FITCP;
    }

    policy_pid = init_log_policy(policy_fd);
    if(INV_HANDLER == policy_pid)
    {
        iret = NOMEM;
        goto FIUDP;
    }

    /*create the daemon thread*/
    iret = pthread_create(&daemon_pid,NULL,hplog_client_daemon,NULL);
    if(iret)
    {
        goto FIPLY;
    }
    
    /*register the signal*/
    signal(SIGTERM,handler_signal);
    /*ignor the sigpipe*/
    signal(SIGPIPE,SIG_IGN);

    pthread_join(policy_pid,NULL);
    pthread_join(daemon_pid,NULL);
    return NOERR;

FIPLY:
    finit_log_policy();
FIUDP:
    finit_protocol_udp();
FITCP:
    finit_protocol_tcp();
FIBKM:
    finit_log_backup();
    #if IPC_SHM
    finitialize_hplog_mutex_sem();
    #endif
FILBM:
    finit_loadbalance();
FICLR:
    finit_log_color();
#if IPC_SHM
FISEM:
    finitialize_share_memory();
#endif
#if IPC_MSG
FITMG:
    msgctl(msg_queue,IPC_RMID,0);
#endif
FSHMM:
    finit_log_doctor();
    return iret;
}

/*test whether the daemon is running?*/
int detecting_daemon_instance()
{
    int    iret,pid,policy_fd;
    struct sockaddr_in serv_addr;

    /*it's time for us to fork my son~~*/
    pid = fork();
    if(pid == -1)
    {
        perror("we don't why failed, but we have the right to complain");
        return FLDON;
    }
    else if(pid)
    {
        /*parent process*/
        /*sleep ns to make sure the daemon is running*/
        usleep(10000);//10 ms
        return NOERR;
    }
    else
    {
        policy_fd = socket(PF_INET,SOCK_DGRAM,0);
        if(policy_fd == -1)
        {
            perror("We must complain\n");
            return FLDON;
        }

        /*fill the address*/
        serv_addr.sin_family = AF_INET;
        serv_addr.sin_port   = htons(HPLOG_CLT_PORT);
        serv_addr.sin_addr.s_addr   = htonl(INADDR_ANY);
        bzero(&(serv_addr.sin_zero),8);

        iret = bind(policy_fd,(struct sockaddr*)&serv_addr,sizeof(struct sockaddr));    
        if(iret == -1)
        {
            close(policy_fd);
            /*the daemon is running*/
            if(errno == EADDRINUSE || errno == EINVAL)
            {
                log_doctor(doctor_level,"The daemon is running,we just need to get the handler\n");
                exit(0); 
            }
            else
            {
                perror("we don't why failed, but we have the right to complain");
                exit(1);
            }
        }
        else
        {      
            setsid();
            /*child process*/
            /*the deamon is down or first laugched by me, start it*/
            initialize_hplog_daemon(policy_fd);
        }
    }
    return NOERR;
}

/* Load the log client daemon
 * 
 * return value:
 * positive means failure,error could be found in the head file or with the help of plogerr();
 * 0 means ok(launched or exist)*/
int load_hplog_client_deamon()
{
    return detecting_daemon_instance();
}


